package jdklearning.concurrent.future;

import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.commons.lang3.time.StopWatch;

import java.lang.management.ManagementFactory;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @author shenenlu 2020年12月23日 下午16:59:58
 */
public class CompletableFutureTest {

    private static ExecutorService threadPool = new ThreadPoolExecutor(10
            , 200
            , 5000
            , TimeUnit.MILLISECONDS
            , new ArrayBlockingQueue<>(10)
            , new BasicThreadFactory.Builder()
                .namingPattern("CompletableFutureTest-thread-pool-%d")
                //守护线程在主线程结束后就会结束
                .daemon(false)
                .priority(Thread.MAX_PRIORITY)
                .build()
            , new ThreadPoolExecutor.CallerRunsPolicy()
//            , new ThreadPoolExecutor.AbortPolicy()
//            , new ThreadPoolExecutor.DiscardOldestPolicy()
//            , new ThreadPoolExecutor.DiscardPolicy()
//                //队列满阻塞
//                , (r, executor) -> {
//                    if (!executor.isShutdown()) {
//                        try {
//                            // 主线程将会被阻塞
//                            executor.getQueue().put(r);
//                        } catch (InterruptedException e) {
//                            // should not be interrupted
//                        }
//                }
    );

    public static Supplier<Integer> supplier1 = () -> {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log("Supplier-步骤1");
        return 6;
    };
    public static Supplier<Integer> supplier2 = () -> {
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log("Supplier-步骤2");
        return 4;
    };
    public static Supplier<Integer> supplierException = () -> {
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log("Supplier-异常步骤");
        return 12/0;
    };



    public static void main(String[] args) throws Exception {
//        CompletableFutureTest completableFutureTest = new CompletableFutureTest();
        String pid = ManagementFactory.getRuntimeMXBean().getName().split("@")[0];
        System.out.println(pid);
        StopWatch sw = new StopWatch();
        sw.start();

//        combineTest();
//        runAyncTest();
//        supplyAsyncTest();
//        thenAcceptBothTest();
//        runAfterBothTest();
//        whenCompleteTest();
//        thenRun();
        thenComposeWhenComplete();

        log("main");

        sw.stop();

        TimeUnit t = TimeUnit.SECONDS;
        System.out.println("time used: " + sw.getTime( t ) + " " + t);

    }


    public static void thenRun(){
        CompletableFuture<Void> future = CompletableFuture.supplyAsync(supplier1)
                .thenRun(() -> System.out.println("World"));
        try {
            future.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    public static void thenComposeWhenComplete(){
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(supplier1, threadPool);
//        CompletableFuture<Integer> future2 = future1.thenCompose(integer -> {
//            CompletableFuture<Integer> completableFuture = new CompletableFuture<>();
//            completableFuture.complete( integer + supplier2.get() );
//            return completableFuture;
//        });

        future1.whenComplete((result, throwable)-> System.out.println(result));


//        try {
//            Thread.sleep(500);
//            Thread.sleep(10000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
    }


    private static void log(Object msg){
        System.out.println(Thread.currentThread().getName()+"-msg = " + msg.toString());
    }


    //无返回值

    /**
     * 没有指定Executor的方法会使用ForkJoinPool.commonPool()
     * 作为它的线程池执行异步代码。
     * 如果指定线程池，则使用指定的线程池运行。以下所有的方法都类同。
     */
    public static void runAsyncTest() throws Exception {
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("run end ...");
        });

        future.get();
    }

    //有返回值
    public static void supplyAsyncTest() throws Exception {
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(supplier1);
        System.out.println("result = " + future.get());
    }


    /***
     * 可以看到Action的类型是BiConsumer<? super T,? super Throwable>它可以处理正常的计算结果，或者异常情况。
     *
     * whenComplete 和 whenCompleteAsync 的区别：
     * whenComplete：是执行当前任务的线程执行继续执行 whenComplete 的任务。
     * whenCompleteAsync：是执行把 whenCompleteAsync 这个任务继续提交给线程池来进行执行。
     */
    public static void whenCompleteTest() throws Exception {
        CompletableFuture<Integer> future =
                CompletableFuture.supplyAsync(supplierException);

        future.whenComplete((integer, throwable) ->
                log("whenCompleteTest done！result = " + integer));

        future.exceptionally(throwable -> {
            log("whenCompleteTest error！");
            return null;
        });
//        TimeUnit.SECONDS.sleep(6);
    }


    /***
     * 当一个线程依赖另一个线程时，可以使用 thenApply 方法来把这两个线程串行化。
     * Function<? super T,? extends U>
     * T：上一个任务返回结果的类型
     * U：当前任务的返回值类型
     */
    public static void thenApplyTest() throws Exception {
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(supplier1);
        future.thenApply((Function<Integer, Object>) integer -> integer * 5);

//        future.join();
        System.out.println(future.get());
    }


    /***
     * 2.3、2个任务进行合并有返回值
     * CompletableFuture.thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
     * 当这个和另一个给定的阶段都正常完成时，两个结果作为提供函数的参数执行
     * CompletableFuture.thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
     * 同thenCombine，新线程执行
     */
    public static void combineTest(){
        CompletableFuture<Integer> integerCompletableFuture =
        CompletableFuture.supplyAsync( supplier1 )
                .thenCombine( CompletableFuture.supplyAsync( supplier2 ), Integer::sum);
        System.out.println(integerCompletableFuture.join());
    }

    /***
     * 2.4 任务合并无返回值
     */
    public static void thenAcceptBothTest(){
        CompletableFuture<Void> integerCompletableFuture =
           CompletableFuture.supplyAsync( supplier1 )
                .thenAcceptBoth( CompletableFuture.supplyAsync( supplier2 ), (one, two) -> System.out.println(one + two));
//        System.out.println(integerCompletableFuture.join());

//        CompletableFuture.supplyAsync(() -> 1)
//                .thenAcceptBoth(CompletableFuture.supplyAsync(() -> 2), (one, two) -> System.out.println(one + two));
    }


    public static void runAfterBothTest(){
//        CompletableFuture<Void> voidCompletableFuture =
//           CompletableFuture.supplyAsync( supplier1 )
//                .thenAcceptBoth( CompletableFuture.supplyAsync( supplier2 ) );


    }
}
